package org.zjt.flink.wordcount;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.JoinOperator;
import org.apache.flink.api.java.operators.ReduceOperator;
import org.apache.flink.api.java.operators.join.JoinFunctionAssigner;
import org.apache.flink.util.Collector;
import org.zjt.flink.wordcount.util.WordCountData;

import java.util.Arrays;
import java.util.List;

/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-09-29 下午5:38
 * @see
 */
public class WordCountQuery {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();


        ReduceOperator<ElementWord> word = executionEnvironment.fromElements(WordCountData.WORDS).flatMap(new TextWordSplit())
                .groupBy("word").reduce((t1, t2) -> {
                    t1.setCnt(t1.getCnt() + t2.getCnt());
                    return t1;
                });

        FlatMapOperator<List<WordMeaning>, WordMeaning> listWordMeaningFlatMapOperator = executionEnvironment.fromElements(Arrays.asList(new WordMeaning("with", "和"), new WordMeaning("sleep", "休眠"), new WordMeaning("sleep", "睡觉")))
                .flatMap(new ElementWordItemSplit());

        JoinOperator.DefaultJoin<ElementWord, WordMeaning> elementWordWordMeaningDefaultJoin = word.join(listWordMeaningFlatMapOperator).where("word").equalTo("word");

        elementWordWordMeaningDefaultJoin.print();
    }


    public static class ElementWordItemSplit implements FlatMapFunction<List<WordMeaning>, WordMeaning> {

        @Override
        public void flatMap(List<WordMeaning> value, Collector<WordMeaning> out) throws Exception {
            value.forEach(out::collect);
        }
    }


    public static class TextWordSplit implements FlatMapFunction<String, ElementWord> {
        @Override
        public void flatMap(String s, Collector<ElementWord> collector) throws Exception {

            String[] tokens = s.toLowerCase().split("\\W+");

            for (String token: tokens) {
                if (token.length() > 0) {
                    ElementWord elementWord = new ElementWord();
                    elementWord.setCnt(1);
                    elementWord.setWord(token);
                    collector.collect(elementWord);
                }
            }
        }
    }


    @Data
    public static class ElementWord{
        private String word;
        private Integer cnt;

    }



    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class WordMeaning{
        private String word;
        private String meaning;
    }
}
